package cn.doitedu.rtmk.demo8.rule_injector;

import org.junit.Test;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class RuleMoni {
    @Test
    public void inject1() throws IOException, SQLException {
        String ruleId = "rule-1-2";

        String ruleParamJson = "{\n" +
                "  \"rule_id\": \"rule-1-2\",\n" +
                "  \"static_profile\": [\n" +
                "    {\n" +
                "      \"tagName\": \"active_level\",\n" +
                "      \"tagValue\": [\n" +
                "        \"2\"\n" +
                "      ],\n" +
                "      \"compareType\": \">\"\n" +
                "    },\n" +
                "    {\n" +
                "      \"tagName\": \"age\",\n" +
                "      \"tagValue\": [\n" +
                "        \"25\",\n" +
                "        \"30\"\n" +
                "      ],\n" +
                "      \"compareType\": \"between\"\n" +
                "    }\n" +
                "  ],\n" +
                "  \"realtime_profile\": {\n" +
                "    \"eventId\": \"g\",\n" +
                "    \"eventCnt\": 2\n" +
                "  },\n" +
                "  \"fire_action\": \"u\"\n" +
                "}";

        // 假装从es查出来符合静态画像条件的人群有  1  3  5
        Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf(1, 2, 5, 7, 9);

        ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
        DataOutputStream dataOut = new DataOutputStream(byteOut);
        bitmap.serialize(dataOut);
        // 规则的预圈选人群bitmap序列化字节
        byte[] bitmapBytes = byteOut.toByteArray();

        // 规则上线状态
        int onlineStatus = 1;

        // 规则的模型运算机代码
        String ruleModelCode = "package cn.doitedu.rtmk.demo8;\n" +
                "\n" +
                "import cn.doitedu.rtmk.common.EventBean;\n" +
                "import com.alibaba.fastjson.JSON;\n" +
                "import com.alibaba.fastjson.JSONObject;\n" +
                "import org.apache.flink.api.common.functions.RuntimeContext;\n" +
                "import org.apache.flink.api.common.state.ValueState;\n" +
                "import org.apache.flink.api.common.state.ValueStateDescriptor;\n" +
                "import org.apache.flink.util.Collector;\n" +
                "import org.roaringbitmap.longlong.Roaring64Bitmap;\n" +
                "\n" +
                "import java.io.IOException;\n" +
                "\n" +
                "/**\n" +
                " * 规则id: rule-1：\n" +
                " * 静态画像条件： age:[20,39] , gender = male\n" +
                " * 实时动态画像标签条件: w行为次数 > 3\n" +
                " * 触发条件: u行为\n" +
                " */\n" +
                "\n" +
                "/**\n" +
                " * 规则id: rule-2：\n" +
                " * 静态画像条件： age:[25,30] , active_level>2\n" +
                " * 实时动态画像标签条件: g行为次数 >=4\n" +
                " * 触发条件: p行为\n" +
                " */\n" +
                "\n" +
                "public class RuleModel_1_Calculator implements RuleCalculator {\n" +
                "\n" +
                "    JSONObject jsonObject;\n" +
                "    ValueState<Integer> rule1State;\n" +
                "\n" +
                "    JSONObject paramJsonObject;\n" +
                "    String ruleId;\n" +
                "    Roaring64Bitmap preSelectUsersBitmap;\n" +
                "\n" +
                "\n" +
                "    @Override\n" +
                "    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap) throws IOException {\n" +
                "\n" +
                "        // 外面注入的 规则的 静态画像条件  预圈选人群\n" +
                "        this.preSelectUsersBitmap = preSelectUsersBitmap;\n" +
                "\n" +
                "        jsonObject = new JSONObject();\n" +
                "\n" +
                "        // 申请一个keyedState，来记录每个用户的规则1的w行为次数\n" +
                "        rule1State = runtimeContext.getState(new ValueStateDescriptor<Integer>(\"rule-1-state\", Integer.class));\n" +
                "\n" +
                "        // 将规则参数，解析成fastjson内部的通用的对象\n" +
                "        paramJsonObject = JSON.parseObject(ruleParamJsonStr);\n" +
                "        ruleId = paramJsonObject.getString(\"rule_id\");\n" +
                "\n" +
                "\n" +
                "    }\n" +
                "\n" +
                "    @Override\n" +
                "    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {\n" +
                "\n" +
                "        String eventId = eventBean.getEventId();\n" +
                "\n" +
                "        // 先判断该行为事件的行为人，如果不属于本规则的目标人群，则直接返回，啥也不做\n" +
                "        if(!preSelectUsersBitmap.contains(eventBean.getUid())) return;\n" +
                "\n" +
                "\n" +
                "        // 0. 判断事件是否是动态画像的目标事件，如果是，则做动态画像统计\n" +
                "        JSONObject realtimeProfileObject = paramJsonObject.getJSONObject(\"realtime_profile\");\n" +
                "        String realtimeProfileEventId = realtimeProfileObject.getString(\"eventId\");\n" +
                "        if (eventId.equals(realtimeProfileEventId)) {\n" +
                "            rule1State.update( ( rule1State.value() == null ? 0 : rule1State.value()) + 1);\n" +
                "        }\n" +
                "\n" +
                "        // 1.判断事件是否是触发条件\n" +
                "        String fireActionEventId = paramJsonObject.getString(\"fire_action\");\n" +
                "        if (eventId.equals(fireActionEventId)) {\n" +
                "            // 2.判断该用户的 “动态画像条件”是否已满足\n" +
                "            Integer realtimeEventCnt = realtimeProfileObject.getInteger(\"eventCnt\");\n" +
                "            if (rule1State.value() != null && rule1State.value() >= realtimeEventCnt) {\n" +
                "\n" +
                "                // 3. 如果上面的比较全部通过，则说明该用户已经命中该规则\n" +
                "                jsonObject.put(\"uid\", eventBean.getUid());\n" +
                "                jsonObject.put(\"timestamp\", eventBean.getTimestamp());\n" +
                "                jsonObject.put(\"rule_id\", ruleId);\n" +
                "\n" +
                "                collector.collect(jsonObject.toJSONString());\n" +
                "\n" +
                "            }\n" +
                "        }\n" +
                "    }\n" +
                "}\n";


        Connection connection = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit40", "root", "root");

        PreparedStatement preparedStatement = connection.prepareStatement("insert into rule_meta values(?,?,?,?,?) ");

        preparedStatement.setString(1, ruleId);
        preparedStatement.setString(2, ruleParamJson);
        preparedStatement.setBytes(3, bitmapBytes);
        preparedStatement.setInt(4, onlineStatus);
        preparedStatement.setString(5, ruleModelCode);

        preparedStatement.execute();


        preparedStatement.close();
        connection.close();


    }


    @Test
    public void inject11() throws IOException, SQLException {
        String ruleId = "rule-1-1";

        String ruleParamJson = "{\n" +
                "  \"rule_id\": \"rule-1-1\",\n" +
                "  \"static_profile\": [\n" +
                "    {\n" +
                "      \"tagName\": \"active_level\",\n" +
                "      \"tagValue\": [\n" +
                "        \"2\"\n" +
                "      ],\n" +
                "      \"compareType\": \">\"\n" +
                "    },\n" +
                "    {\n" +
                "      \"tagName\": \"age\",\n" +
                "      \"tagValue\": [\n" +
                "        \"25\",\n" +
                "        \"30\"\n" +
                "      ],\n" +
                "      \"compareType\": \"between\"\n" +
                "    }\n" +
                "  ],\n" +
                "  \"realtime_profile\": {\n" +
                "    \"eventId\": \"y\",\n" +
                "    \"eventCnt\": 3\n" +
                "  },\n" +
                "  \"fire_action\": \"t\"\n" +
                "}";

        // 假装从es查出来符合静态画像条件的人群有  1  3  5
        Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf(1, 4, 5, 8);

        ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
        DataOutputStream dataOut = new DataOutputStream(byteOut);
        bitmap.serialize(dataOut);
        // 规则的预圈选人群bitmap序列化字节
        byte[] bitmapBytes = byteOut.toByteArray();

        // 规则上线状态
        int onlineStatus = 1;

        // 规则的模型运算机代码
        String ruleModelCode = "package cn.doitedu.rtmk.demo8;\n" +
                "\n" +
                "import cn.doitedu.rtmk.common.EventBean;\n" +
                "import com.alibaba.fastjson.JSON;\n" +
                "import com.alibaba.fastjson.JSONObject;\n" +
                "import org.apache.flink.api.common.functions.RuntimeContext;\n" +
                "import org.apache.flink.api.common.state.ValueState;\n" +
                "import org.apache.flink.api.common.state.ValueStateDescriptor;\n" +
                "import org.apache.flink.util.Collector;\n" +
                "import org.roaringbitmap.longlong.Roaring64Bitmap;\n" +
                "\n" +
                "import java.io.IOException;\n" +
                "\n" +
                "/**\n" +
                " * 规则id: rule-1：\n" +
                " * 静态画像条件： age:[20,39] , gender = male\n" +
                " * 实时动态画像标签条件: w行为次数 > 3\n" +
                " * 触发条件: u行为\n" +
                " */\n" +
                "\n" +
                "/**\n" +
                " * 规则id: rule-2：\n" +
                " * 静态画像条件： age:[25,30] , active_level>2\n" +
                " * 实时动态画像标签条件: g行为次数 >=4\n" +
                " * 触发条件: p行为\n" +
                " */\n" +
                "\n" +
                "public class RuleModel_1_Calculator implements RuleCalculator {\n" +
                "\n" +
                "    JSONObject jsonObject;\n" +
                "    ValueState<Integer> rule1State;\n" +
                "\n" +
                "    JSONObject paramJsonObject;\n" +
                "    String ruleId;\n" +
                "    Roaring64Bitmap preSelectUsersBitmap;\n" +
                "\n" +
                "\n" +
                "    @Override\n" +
                "    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap) throws IOException {\n" +
                "\n" +
                "        // 外面注入的 规则的 静态画像条件  预圈选人群\n" +
                "        this.preSelectUsersBitmap = preSelectUsersBitmap;\n" +
                "\n" +
                "        jsonObject = new JSONObject();\n" +
                "\n" +
                "        // 申请一个keyedState，来记录每个用户的规则1的w行为次数\n" +
                "        rule1State = runtimeContext.getState(new ValueStateDescriptor<Integer>(\"rule-1-state\", Integer.class));\n" +
                "\n" +
                "        // 将规则参数，解析成fastjson内部的通用的对象\n" +
                "        paramJsonObject = JSON.parseObject(ruleParamJsonStr);\n" +
                "        ruleId = paramJsonObject.getString(\"rule_id\");\n" +
                "\n" +
                "\n" +
                "    }\n" +
                "\n" +
                "    @Override\n" +
                "    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {\n" +
                "\n" +
                "        String eventId = eventBean.getEventId();\n" +
                "\n" +
                "        // 先判断该行为事件的行为人，如果不属于本规则的目标人群，则直接返回，啥也不做\n" +
                "        if(!preSelectUsersBitmap.contains(eventBean.getUid())) return;\n" +
                "\n" +
                "\n" +
                "        // 0. 判断事件是否是动态画像的目标事件，如果是，则做动态画像统计\n" +
                "        JSONObject realtimeProfileObject = paramJsonObject.getJSONObject(\"realtime_profile\");\n" +
                "        String realtimeProfileEventId = realtimeProfileObject.getString(\"eventId\");\n" +
                "        if (eventId.equals(realtimeProfileEventId)) {\n" +
                "            rule1State.update( ( rule1State.value() == null ? 0 : rule1State.value()) + 1);\n" +
                "        }\n" +
                "\n" +
                "        // 1.判断事件是否是触发条件\n" +
                "        String fireActionEventId = paramJsonObject.getString(\"fire_action\");\n" +
                "        if (eventId.equals(fireActionEventId)) {\n" +
                "            // 2.判断该用户的 “动态画像条件”是否已满足\n" +
                "            Integer realtimeEventCnt = realtimeProfileObject.getInteger(\"eventCnt\");\n" +
                "            if (rule1State.value() != null && rule1State.value() >= realtimeEventCnt) {\n" +
                "\n" +
                "                // 3. 如果上面的比较全部通过，则说明该用户已经命中该规则\n" +
                "                jsonObject.put(\"uid\", eventBean.getUid());\n" +
                "                jsonObject.put(\"timestamp\", eventBean.getTimestamp());\n" +
                "                jsonObject.put(\"rule_id\", ruleId);\n" +
                "\n" +
                "                collector.collect(jsonObject.toJSONString());\n" +
                "\n" +
                "            }\n" +
                "        }\n" +
                "    }\n" +
                "}\n";


        Connection connection = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit40", "root", "root");

        PreparedStatement preparedStatement = connection.prepareStatement("insert into rule_meta values(?,?,?,?,?) ");

        preparedStatement.setString(1, ruleId);
        preparedStatement.setString(2, ruleParamJson);
        preparedStatement.setBytes(3, bitmapBytes);
        preparedStatement.setInt(4, onlineStatus);
        preparedStatement.setString(5, ruleModelCode);

        preparedStatement.execute();


        preparedStatement.close();
        connection.close();


    }


    @Test
    public void inject2() throws Exception{

        String ruleId = "rule-2-1";
        String ruleParamJson = "{\n" +
                " \"rule_id\":\"rule-2-1\",\n" +
                " \"event_id\":\"e\"\n" +
                "}";
        String ruleModelCode = "package cn.doitedu.rtmk.demo8;\n" +
                "\n" +
                "import cn.doitedu.rtmk.common.EventBean;\n" +
                "import com.alibaba.fastjson.JSON;\n" +
                "import com.alibaba.fastjson.JSONObject;\n" +
                "import org.apache.flink.api.common.functions.RuntimeContext;\n" +
                "import org.apache.flink.util.Collector;\n" +
                "import org.roaringbitmap.longlong.Roaring64Bitmap;\n" +
                "\n" +
                "import java.io.IOException;\n" +
                "\n" +
                "/**\n" +
                "{\n" +
                " \"rule_id\":\"rule-2-1\"\n" +
                " \"event_id\":\"e\"\n" +
                "}\n" +
                " */\n" +
                "public class RuleModel2Calculator implements RuleCalculator{\n" +
                "\n" +
                "    Roaring64Bitmap preSelectUsersBitmap;\n" +
                "    String eventId;\n" +
                "    String ruleId;\n" +
                "    JSONObject message;\n" +
                "    \n" +
                "    @Override\n" +
                "    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap) throws IOException {\n" +
                "        this.preSelectUsersBitmap = preSelectUsersBitmap;\n" +
                "        JSONObject jsonObject = JSON.parseObject(ruleParamJsonStr);\n" +
                "        eventId = jsonObject.getString(\"event_id\");\n" +
                "        ruleId = jsonObject.getString(\"rule_id\");\n" +
                "\n" +
                "        message = new JSONObject();\n" +
                "    }\n" +
                "\n" +
                "    @Override\n" +
                "    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {\n" +
                "\n" +
                "        if(preSelectUsersBitmap.contains(eventBean.getUid())  && eventId.equals(eventBean.getEventId())){\n" +
                "            message.put(\"rule_id\",ruleId);\n" +
                "            message.put(\"user_id\",eventBean.getUid());\n" +
                "            message.put(\"timestamp\",eventBean.getTimestamp());\n" +
                "            collector.collect(message.toJSONString());\n" +
                "        }\n" +
                "        \n" +
                "        \n" +
                "    }\n" +
                "}\n";


        int onlineStatus = 1;

        Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf(1, 3, 6);
        ByteArrayOutputStream bout = new ByteArrayOutputStream();
        DataOutputStream dout = new DataOutputStream(bout);
        bitmap.serialize(dout);
        byte[] bitmapBytes = bout.toByteArray();


        Connection connection = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit40", "root", "root");

        PreparedStatement preparedStatement = connection.prepareStatement("insert into rule_meta values(?,?,?,?,?) ");

        preparedStatement.setString(1, ruleId);
        preparedStatement.setString(2, ruleParamJson);
        preparedStatement.setBytes(3, bitmapBytes);
        preparedStatement.setInt(4, onlineStatus);
        preparedStatement.setString(5, ruleModelCode);

        preparedStatement.execute();


        preparedStatement.close();
        connection.close();



    }

}
